package sheep.kangzz

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ListBuffer

/**
 * 找相同用户
 * Author: sheep.Old
 * QQ: 64341393
 * Date: 2018/5/17
 */
object Graph4CommonUsers {

    def main(args: Array[String]): Unit = {

        val sparkConf = new SparkConf()
            .setAppName(this.getClass.getSimpleName)
            .setMaster("local[*]")
            .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

        val sc = new SparkContext(sparkConf)

        // 读取数据
        val log = sc.textFile("/Users/Jiawei/Downloads/source_data_2012_05.data.part")


        val filtered = log.map(line => line.split("\t")(1))
            .filter(line => "\"prev_out\"".r.findFirstIn(line).nonEmpty && "\"addr\"".r.findFirstIn(line).nonEmpty)
            .map(JSON.parseObject(_))


        filtered.map(line => {

        })


        filtered.mapPartitions(part => {

            val list = new ListBuffer[(String, Int)]()

            part.foreach(line => {
                val json = JSON.parseObject(line)

                val hash = json.getString("hash")
                val addArrays = json.getJSONArray("inputs")

                for (i <- 0 until addArrays.size()) {
                    val addr = addArrays.getJSONObject(i).getJSONObject("prev_out").getString("addr")
                    list.append((addr, hash.hashCode))
                }
            })
            list.iterator
        })





        sc.stop()
    }


}
